/**
 * Copyright 2018 asiainfo Inc.
 **/
package com.myz.rabbitmqspring.spring.event.service;

import com.myz.rabbitmqspring.spring.event.dto.EventMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author maoyz on 2018/8/28
 * @version: v1.0
 */
@Slf4j
public class MessageAdapterHandler {

    private ConcurrentHashMap<String, EventProcessorWrap> epwMap;

    public MessageAdapterHandler() {
        this.epwMap = new ConcurrentHashMap<>();
    }

    public void handleMessage(EventMessage eventMessage) {
        log.debug("Receive an EventMessage: [" + eventMessage + "]");
        // 先要判断接收到的message是否是空的，在某些异常情况下，会产生空值
        if (eventMessage == null) {
            log.warn("Receive an null EventMessage, it may product some errors, and processing message is canceled.");
            return;
        }
        if (StringUtils.isEmpty(eventMessage.getQueueName()) || StringUtils.isEmpty(eventMessage.getExchangeName())) {
            log.warn("The EventMessage's queueName and exchangeName is empty, this is not allowed, and processing message is canceled.");
            return;
        }
        // 解码，并交给对应的EventHandle执行
        EventProcessorWrap eepw = epwMap.get(eventMessage.getQueueName() + "|" + eventMessage.getExchangeName());
        if (eepw == null) {
            log.warn("Receive an EopEventMessage, but no processor can do it.");
            return;
        }
        try {
            eepw.process(eventMessage.getEventData());
        } catch (IOException e) {
            log.error("Event content can not be Deserialized, check the provided CodecFactory.", e);
            return;
        }
    }

    protected void add(String queueName, String exchangeName, EventProcessor processor, CodecFactory codecFactory) {
        if (StringUtils.isEmpty(queueName) || StringUtils.isEmpty(exchangeName) || processor == null || codecFactory == null) {
            throw new RuntimeException("queueName and exchangeName can not be empty,and processor or codecFactory can not be null. ");
        }
        EventProcessorWrap epw = new EventProcessorWrap(codecFactory, processor);
        EventProcessorWrap oldProcessorWrap = epwMap.putIfAbsent(queueName + "|" + exchangeName, epw);
        if (oldProcessorWrap != null) {
            log.warn("The processor of this queue and exchange exists, and the new one can't be add");
        }
    }

    protected Set<String> getAllBinding() {
        Set<String> keySet = epwMap.keySet();
        return keySet;
    }

    protected static class EventProcessorWrap {

        private CodecFactory codecFactory;

        private EventProcessor eep;

        protected EventProcessorWrap(CodecFactory codecFactory,
                                     EventProcessor eep) {
            this.codecFactory = codecFactory;
            this.eep = eep;
        }

        public void process(byte[] eventData) throws IOException {
            Object obj = codecFactory.deSerialize(eventData);
            eep.process(obj);
        }
    }

}
